﻿// See https://aka.ms/new-console-template for more information

// Stream configuration
using Streamiz.Kafka.Net.SerDes;
using Streamiz.Kafka.Net.Stream;
using Streamiz.Kafka.Net;
using Streamiz.Kafka.Net.Table;
using Confluent.Kafka;
using Streamiz.Kafka.Net.Metrics;
using Confluent.Kafka.Admin;
using Microsoft.Extensions.Logging;
using GEG.NGCRM.Aggregate.Host.Test;
using System.Text;
using Streamiz.Kafka.Net.SchemaRegistry.SerDes.Json;
using Newtonsoft.Json.Linq;


var config = new StreamConfig<StringSerDes, StringSerDes>
{
    ApplicationId = $"test-app",
    BootstrapServers = "192.168.50.69:9092",
    Acks = Acks.All,
    AutoOffsetReset = AutoOffsetReset.Earliest,
    Logger = LoggerFactory.Create((b) =>
    {
        b.AddConsole();
        b.SetMinimumLevel(LogLevel.Information);
    }),
    FollowMetadata = true,
    DefaultKeySerDes = new StringSerDes(),
    DefaultValueSerDes = new JsonSerDes<AggregateEventData>(),
    SecurityProtocol = SecurityProtocol.Plaintext,
};



var stringSerdes = new StringSerDes();
var messageSerdes = new JsonSerDes<AggregateEventData>();

var builder = new StreamBuilder();
var stream1 = builder.Stream<string, AggregateEventData>(nameof(AggregateType.HotelBooking), stringSerdes, messageSerdes);

var stream2 = builder.Stream<string, AggregateEventData>(nameof(AggregateType.POSExpenses), stringSerdes, messageSerdes);

var messageStream = stream1.Merge(stream2);

messageStream
                .Map((k, v, _) => KeyValuePair.Create(v.Guest_Id, v))
                .GroupByKey()
                //.WindowedBy(TumblingWindowOptions.Of(200000))
                .Aggregate<AggregateSummaryEventData, JsonSerDes<AggregateSummaryEventData>>(
                    () =>
                    {
                        var aggregateSummaryEventData = new AggregateSummaryEventData
                        {

                        };
                        return aggregateSummaryEventData;

                    },
                     (key, aggregateEventData, aggregateSummaryEventData) =>
                    {
                        var now = DateTime.Now;
                        var nowScope = now.Date.AddDays((-365 * 2 + 1));

                        aggregateSummaryEventData.Guest_Id = key;
                        aggregateSummaryEventData.Update_Dtm = now;

                        if (aggregateEventData.Type == AggregateType.HotelBooking)
                        {
                            var metrics = aggregateSummaryEventData.HotelBooking_Metrics.Where(a => a.Dtm.Date == aggregateEventData.Dtm.Date).FirstOrDefault();
                            if (metrics != null)
                            {
                                metrics.Amount += aggregateEventData.Billing_Amount;
                            }
                            else
                            {
                                metrics = new AggregateSummaryEventData_Metric()
                                {
                                    Amount = aggregateEventData.Billing_Amount,
                                    Dtm = aggregateEventData.Dtm.Date,
                                };
                                aggregateSummaryEventData.HotelBooking_Metrics.Add(metrics);
                            }
                        }
                        aggregateSummaryEventData.HotelBooking_Metrics = aggregateSummaryEventData.HotelBooking_Metrics.Where(a => a.Dtm > nowScope).OrderBy(a => a.Dtm).ToList();

                        if (aggregateEventData.Type == AggregateType.POSExpenses)
                        {
                            var metrics = aggregateSummaryEventData.POSExpenses_Metrics.Where(a => a.Dtm.Date == aggregateEventData.Dtm.Date).FirstOrDefault();
                            if (metrics != null)
                            {
                                metrics.Amount += aggregateEventData.Billing_Amount;
                            }
                            else
                            {
                                metrics = new AggregateSummaryEventData_Metric()
                                {
                                    Amount = aggregateEventData.Billing_Amount,
                                    Dtm = aggregateEventData.Dtm.Date,
                                };
                                aggregateSummaryEventData.POSExpenses_Metrics.Add(metrics);
                            }
                        }
                        aggregateSummaryEventData.POSExpenses_Metrics = aggregateSummaryEventData.POSExpenses_Metrics.Where(a => a.Dtm > nowScope).OrderBy(a => a.Dtm).ToList();
                        return aggregateSummaryEventData;
                    }
                )
                .ToStream()
                .Map((k, v, c) =>
                {
                    return KeyValuePair.Create(k, v);
                })
                .To<StringSerDes, JsonSerDes<AggregateSummaryEventData>>("Test");



var topology = builder.Build();

var stream = new KafkaStream(topology, config);

Console.CancelKeyPress += (_, _) => { stream.Dispose(); };

await stream.StartAsync();